#!/usr/env python
# Rob Dennis - Arclite EMP
# TVDB API
# Project Homepage: http://code.google.com/p/keasby/

#--- Python Imports
import collections
import logging
import os
import sys
import urllib
import time
import zipfile

from contextlib import closing
from os.path import join as pjoin
from xml.etree import ElementTree as ET

#--- TVDB API Imports
from api import *

log = logging.getLogger("tvdb.unittests")

#Special handling for pre-python2.7 unittests
if sys.version_info.major == 2 and sys.version_info.minor >= 7:
    log.debug("loading 2.7 and later unittest")
    import unittest
else:
    import unittest2 as unittest

class TvDbApiTestCase(unittest.TestCase):
    """
    Tests the use of the TvDbApi class
    """

    parent_test_dir = os.path.join(os.path.dirname(__file__), "test")
    test_dir = os.path.join(os.path.dirname(__file__), "test", "tvdb")
    test_assets = os.path.join(os.path.dirname(__file__), "test", "tvdb.zip")

    _TvDbTestItem = collections.namedtuple("_TvDbTestItem",
                                        ["series",
                                        "episode",
                                        "paths"])

    class TvDbSchema(collections.MutableMapping):
        def __contains__(self, key):
            return key in self.__dict__ and self.__dict__[key] is not None
        def __len__(self):
            return len(self.__dict__)
        def __setitem__(self, key, value):
            self.__dict__[key] = value
            return self
        def __getitem__(self, key):
            return self.__dict__[key]
        def __delitem__(self, key):
            del self.__dict__[key]
        def __iter__(self):
            return self.__dict__.__iter__()
        def __str__(self):
            return " ".join(["%s=%s"%(key, value) for key, value
                                in self.__dict__.iteritems()])
        def __repr__(self):
            return "<%s: %s>"%(self.__class__.__name__, self.__str__)


    class TvDbSeries(TvDbSchema):
        def __init__(self, **kwargs):
            self.__dict__ =  dict.fromkeys(["id",
                                            "Actors",
                                            "Airs_DayOfWeek",
                                            "Airs_Time",
                                            "ContentRating",
                                            "FirstAired",
                                            "Genre",
                                            "IMDB_ID",
                                            "Language",
                                            "Network",
                                            "NetworkID",
                                            "Overview",
                                            "Rating",
                                            "Runtime",
                                            "SeriesID",
                                            "SeriesName",
                                            "Status",
                                            "added",
                                            "addedBy",
                                            "banner",
                                            "fanart",
                                            "lastupdated",
                                            "poster",
                                            "zap2it_id"])
            self.__dict__.update(kwargs)



    class TvDbEpisode(TvDbSchema):
        def __init__(self, **kwargs):

            self.__dict__ =  dict.fromkeys(["id",
                                            "Combined_episodenumber",
                                            "Combined_season",
                                            "DVD_chapter",
                                            "DVD_discid",
                                            "DVD_episodenumber",
                                            "DVD_season",
                                            "Director",
                                            "EpImgFlag",
                                            "EpisodeName",
                                            "EpisodeNumber",
                                            "FirstAired",
                                            "GuestStars",
                                            "IMDB_ID",
                                            "Language",
                                            "Overview",
                                            "ProductionCode",
                                            "Rating",
                                            "SeasonNumber",
                                            "Writer",
                                            "absolute_number",
                                            "filename",
                                            "lastupdated",
                                            "seasonid",
                                            "seriesid"])
            self.__dict__.update(kwargs)


    #A default episode to test against
    test_def = _TvDbTestItem(TvDbSeries(SeriesName="Lost", id=73739),
                             TvDbEpisode(id=127131,
                                         EpisodeName="Pilot (1)",
                                         SeasonNumber=1,
                                         EpisodeNumber=1),
                             dict(fsr_zip=pjoin(test_dir,
                                                    "series",
                                                    "73739",
                                                    "all",
                                                    "en.zip"),
                                  fsr_xml=pjoin(test_dir,
                                                    "series",
                                                    "73739",
                                                    "all",
                                                    "en.xml"),
                                  bsr_xml=pjoin(test_dir,
                                                    "series",
                                                    "73739",
                                                    "en.xml"),
                                  ber_xml=pjoin(test_dir,
                                                    "episodes",
                                                    "127131",
                                                    "en.xml"),
                                  query_xml=pjoin(test_dir,
                                                   "query_GetSeries-Lost.xml")))




    #Not all shows have a mismatch between airing (default) and DVD order
    #this is one
    test_dvd = _TvDbTestItem(TvDbSeries(SeriesName="Firefly", id=78874),
                             TvDbEpisode(id=297999,
                                         EpisodeName="Serenity",
                                         SeasonNumber=1,
                                         EpisodeNumber=11,
                                         DVD_season=1,
                                         DVD_episodenumber=1.0),
                             dict(fsr_zip=pjoin(test_dir,
                                                    "series",
                                                    "78874",
                                                    "all",
                                                    "en.zip"),
                                  fsr_xml=pjoin(test_dir,
                                                    "series",
                                                    "78874",
                                                    "all",
                                                    "en.xml"),
                                  bsr_xml=pjoin(test_dir,
                                                    "series",
                                                    "78874",
                                                    "en.xml"),
                                  ber_xml=pjoin(test_dir,
                                                    "episodes",
                                                    "297999",
                                                    "en.xml"),
                                  query_xml=pjoin(test_dir,
                                                "query_GetSeries-Firefly.xml")))

    #Not all shows have absolutenumber present, so here's a test for that:
    #Bleach is the example from: http://thetvdb.com/wiki/index.php/abs_order
    test_abs = _TvDbTestItem(TvDbSeries(SeriesName="Bleach", id=74796),
                             TvDbEpisode(id=155071, EpisodeName="The Day I "+
                                                           "Became a Shinigami",
                                                    absolute_number=1),
                            dict(fsr_zip=pjoin(test_dir,
                                                    "series",
                                                    "74796",
                                                    "all",
                                                    "en.zip"),
                                  fsr_xml=pjoin(test_dir,
                                                    "series",
                                                    "74796",
                                                    "all",
                                                    "en.xml"),
                                  query_xml=pjoin(test_dir,
                                                 "query_GetSeries-Bleach.xml")))

    test_all = [test_def, test_dvd, test_abs]
    test_seriesids = [test.series.id for test in test_all]

    test_access_battery = (((test_def.series.SeriesName,
                             (test_def.episode.SeasonNumber,
                             test_def.episode.EpisodeNumber),
                             TvDbApi.def_order),
                             test_def.episode.EpisodeName),
                            ((test_def.series.SeriesName,
                             (test_def.episode.SeasonNumber,
                             test_def.episode.EpisodeNumber)),
                             test_def.episode.EpisodeName),
                            ((test_dvd.series.SeriesName,
                             (test_dvd.episode.DVD_season,
                             test_dvd.episode.DVD_episodenumber),
                             TvDbApi.dvd_order),
                             test_dvd.episode.EpisodeName),
                            ((test_abs.series.SeriesName,
                             test_abs.episode.absolute_number,
                             TvDbApi.abs_order),
                             test_abs.episode.EpisodeName))                                 
    
    test_seriesid_access_battery = (((test_def.series.id,
                                     (test_def.episode.SeasonNumber,
                                     test_def.episode.EpisodeNumber),
                                     TvDbApi.def_order),
                                     test_def.episode.EpisodeName),
                                    ((test_def.series.id,
                                     (test_def.episode.SeasonNumber,
                                     test_def.episode.EpisodeNumber)),
                                     test_def.episode.EpisodeName),
                                    ((test_dvd.series.id,
                                     (test_dvd.episode.DVD_season,
                                     test_dvd.episode.DVD_episodenumber),
                                     TvDbApi.dvd_order),
                                     test_dvd.episode.EpisodeName),
                                    ((test_abs.series.id,
                                     test_abs.episode.absolute_number,
                                     TvDbApi.abs_order),
                                     test_abs.episode.EpisodeName))
                                     
    
    #selecting a show with a basic enough name to ensure several options
    test_query_sid = test_def.episode.id
    test_query_name = test_def.series.SeriesName
    test_query_lang = u"en.xml"

    def setUp(self):
        """
        instantiates a TvDbApi instance
        """

        self.assertTrue(os.path.exists(self.test_assets),
                        "%s does not exist"%self.test_assets)

        try:
            zipfile.ZipFile(self.test_assets,
                            "r").extractall(self.parent_test_dir)
        except Exception, msg:
            log.warning("issue extracting test files: %s", msg)


        self.api = TvDbApi("D1EBA6781E2572BB", bind='sqlite:///:memory:',
                            cache=self.test_dir)

    def tearDown(self):
        """
        cleans up and prepares for next test
        """
        #necessary only for unit tests
        import shutil

        self.api.purge(True)

        shutil.rmtree(self.test_dir, ignore_errors=True)

        del self.api

    def test_mirrors(self):
        """
        tests that there exists a mirror for each type of TVDB resource
        """

        self.assertTrue(self.api.get_mirror("xml"))
        self.assertTrue(self.api.get_mirror("zip"))
        self.assertTrue(self.api.get_mirror("banner"))
        self.failUnlessRaises(TvDbApiError, self.api.get_mirror,
                                            "invalid_mirror")

    def test_queries(self):
        """
        test the results of queries available from thetvdb.com
        """
        test_time = str(time.time() - 60*60*24)

        queries = []
        #normal GetSeries, TvDb defaults to english
        queries.append((self.api.query_GetSeries,
                        (self.test_query_name,),
                        dict()))
        #GetSeries with specified language
        queries.append((self.api.query_GetSeries,
                       (self.test_query_name, self.test_query_lang),
                       dict()))
        #GetSeries with "all"
        queries.append((self.api.query_GetSeries,
                       (self.test_query_name, u"all"),
                       dict()))
        #return current time
        queries.append((self.api.query_Updates,
                        tuple(),
                        dict(type="none")))
        #return current epoch time
        queries.append((self.api.query_Updates,
                        tuple(),
                        dict(type="none")))
        #return updated series (unspecified type)
        queries.append((self.api.query_Updates,
                        tuple(),
                        dict(time=test_time)))
        #return updated series
        queries.append((self.api.query_Updates,
                        (test_time, "series"),
                        dict()))
        #return updated series in unicode
        queries.append((self.api.query_Updates,
                        (test_time, u"series"),
                        dict()))
        #return updated episodes
        queries.append((self.api.query_Updates,
                        (test_time, "episode"),
                        dict()))
        #return updated series
        queries.append((self.api.query_Updates,
                        (test_time, "all"),
                        dict()))
        #return epoch time (with extraneous time parameter)
        queries.append((self.api.query_Updates,
                        (test_time, "none"),
                        dict()))

        for api_query, args, kwargs in queries:
            try:
                with closing(urllib.urlopen(api_query(*args,
                                                      **kwargs))) as xml_result:
                    parsed_xml = ET.parse(xml_result)
                    self.assertTrue(parsed_xml,
                                    "%s gives invalid data"%xml_result.geturl())
            except IOError, msg:
                log.warning("non test error with: %s, %s, %s", api_query,
                                                                args,
                                                                kwargs)
                log.warning(msg)

    def test_catching_invalid_queries(self):
        """
        test to ensure that invalid queries raise appropriate exceptions
        """

        test_time = str(time.time() - 60*60*24)

        #invalid query state
        self.failUnlessRaises(TvDbApiError, self.api.query_Updates)
        #invalid type
        self.failUnlessRaises(TvDbApiError, self.api.query_Updates,
                                            None, "invalid")
        #invalid type with time
        self.failUnlessRaises(TvDbApiError, self.api.query_Updates,
                                            test_time, "invalid")

    def test_add_series(self):
        """
        Test adding a series to the list and confirming that series\
        added is remembered
        """

        self.api.add_series(*self.test_seriesids[:-1])
        self.api.add_series(self.test_seriesids[-1])

        for seriesid in self.test_seriesids:
            self.assertTrue(seriesid in self.api.seriesid_list,
                            "%d not added successfully"%seriesid)

    def test_update_series(self):
        """
        Test the method in which series information is updated/stored
        """

        self.api.add_series(*self.test_seriesids)
        self.api.update()

    def test_update_series_with_manual_seriesid(self):
        """
        Test the method in which series information is updated/stored\
        when certain seriesids are added a different seriesid is updated
        """

        self.api.add_series(*self.test_seriesids[:-1])
        self.api.update(self.test_seriesids[-1])

    def test_update_series_with_manual_seriesid_and_no_initial_add(self):
        """
        Test the method in which series information is updated/stored\
        when no seriesids are added a different seriesid is updated
        """

        self.api.update(*self.test_seriesids)

    def test_update_series_incrementally(self):
        """
        Test the method in which series information is updated/stored\
        when no seriesids are added and multiple updates are done in series
        """

        for seriesid in self.test_seriesids:
            self.api.update(seriesid)

    def test_update_series_and_confirm_results(self):
        """
        Test the method to update a series and test information again\
        known values
        """

        for test in self.test_all:
            self.api.update(test.series.id)

            series = TvDbApi.Series.get_one(id=test.series.id)

            self.assertEquals(test.series.SeriesName, series.SeriesName)
            self.assertEquals(test.series.id, series.id)

            episode = TvDbApi.Episode.get_one(id=test.episode.id)

            self.assertEquals(test.episode.id, episode.id)
            self.assertEquals(test.episode.EpisodeName, episode.EpisodeName)

    def test_map_series_name_to_seriesid(self):
        """
        test the ability to search a series name, map it to a seriesid, and\
        have that selection persist.
        """

        for test in self.test_all:

            name = test.series.SeriesName

            search_gen = ((seriesid == test.series.id)
                            for seriesid
                            in self.api.search(name))

            self.assertTrue(any(search_gen))
            self.assertEquals(TvDbApi.SearchMap.get_one(name=name).id,
                                test.series.id)
                                
    def test_find_after_update(self):
        """
        test the ability to get an episode from series name with an update
        """
        
        self.api.update(*self.test_seriesids)
        
        for find_args, test_name in self.test_access_battery:
            found_name = self.api.find(*find_args).EpisodeName
            self.assertEquals(test_name,
                              found_name,
                              "for %s, %s!=%s"%(str(find_args),
                                                test_name,
                                                found_name))
                                                
    def test_find_without_update(self):
        """
        test the ability to get an episode from series name without an update
        """
        
        for find_args, test_name in self.test_access_battery:
            found_name = self.api.find(*find_args).EpisodeName
            self.assertEquals(test_name,
                              found_name,
                              "for %s, %s!=%s"%(str(find_args),
                                                test_name,
                                                found_name))
                                                
    def test_find_with_seriesids_and_update(self):
        """
        test the ability to get an episode from seriesid without an update
        """
        
        self.api.update(*self.test_seriesids)
        
        for find_args, test_name in self.test_seriesid_access_battery:
            found_name = self.api.find(*find_args).EpisodeName
            self.assertEquals(test_name,
                              found_name,
                              "for %s, %s!=%s"%(str(find_args),
                                                test_name,
                                                found_name))

    def test_find_with_seriesids_and_no_update(self):
        """
        test the ability to get an episode from seriesid without an update
        """
        
        for find_args, test_name in self.test_seriesid_access_battery:
            found_name = self.api.find(*find_args).EpisodeName
            self.assertEquals(test_name,
                              found_name,
                              "for %s, %s!=%s"%(str(find_args),
                                                test_name,
                                                found_name))

    def test_find_using_getitem_syntax(self):
        """
        test the ability to get episodes using get item (bracket syntax)
        """
        
        for find_args, test_name in self.test_access_battery:
            if len(find_args) == 3:
                name, numbers, order = find_args
                if isinstance(numbers, int):
                    abs_number = numbers
                    found_episode = self.api[order][name][abs_number]
                else:
                    season, episode = numbers
                    found_episode = self.api[order][name][season][episode]
                

            else:
                name, numbers = find_args
                season, episode = numbers
                found_episode = self.api[name][season][episode]
                
            found_name = found_episode.EpisodeName
            self.assertEquals(test_name,
                              found_name,
                              "for %s, %s!=%s"%(str(find_args),
                                                test_name,
                                                found_name))
                
if __name__ == "__main__":
    suite = unittest.TestSuite()
    suite.addTest(unittest.makeSuite(TvDbApiTestCase))
    unittest.TextTestRunner(verbosity=2).run(suite)